using System;
using System.Collections.Generic;
using System.Text;
using System.Linq;
using SimpleServiceBus.Bus.Transport;
using SimpleServiceBus.Utilities;
using System.Collections.Specialized;
using System.Net;
using System.Globalization;
using Apache.NMS;
using SimpleServiceBus.Bus;
using Apache.NMS.ActiveMQ.Commands;
using System.Threading;
using System.Diagnostics;
using Apache.NMS.ActiveMQ;
using SimpleServiceBus.Bus.MessageManagement.Serialization;
using SimpleServiceBus.Bus.Pipeline;
using SimpleServiceBus.Bus.MessageManagement;
using SimpleServiceBus.Endpoint;
using SimpleServiceBus.Utilities.ObjectBuilder;
using System.Xml.Serialization;
using System.ComponentModel;
using System.Collections;
using Common.Logging;

namespace SimpleServiceBus.Endpoints.ActiveMQ.Transport {

    public class ActiveMQTransport: EndpointServiceBase, ITransport, IMessageProcessingErrorHandler {

        //-----< Fields >-----
        //----------------------------------------------------------------------------------        
        public const int DefaultActiveMQPort = 61616;
        private static HashSet<String> fExcludedHeaderProperties;

        private Object fConnectionLock = new Object();
        private IConnection fConnection;
        private IDestinationResolver fDestinationResolver;
        private ActiveMQConsumerPoolCollection fConsumerPools = new ActiveMQConsumerPoolCollection();

        //-----< Constructors >-----
        //----------------------------------------------------------------------------------        
        static ActiveMQTransport() {
            fExcludedHeaderProperties = new HashSet<string>();
            fExcludedHeaderProperties.Add(MessageEnvelope.MessageID_HeaderKey);
            fExcludedHeaderProperties.Add(MessageEnvelope.CorrelationID_HeaderKey);
            fExcludedHeaderProperties.Add(MessageEnvelope.Recoverable_HeaderKey);
            fExcludedHeaderProperties.Add(MessageEnvelope.RetryCount_HeaderKey);
            fExcludedHeaderProperties.Add(MessageEnvelope.ReturnAddress_HeaderKey);
        }

        public ActiveMQTransport() {
        }

        //-----< Public and protected properties >-----
        //----------------------------------------------------------------------------------        

        public IConnectionFactory ConnectionFactory { get; set; }

        public ActiveMQConsumerPoolCollection ConsumerPools {
            get { return fConsumerPools; }
        }

        public IDestinationResolver DestinationResolver {
            get { return fDestinationResolver; }
            set {
                if(value == null)
                    throw new ArgumentNullException(String.Format(CultureInfo.InvariantCulture, "Cannot clear the DestinationResolver property"), "value");
                fDestinationResolver = value;    
            }
        }

        public virtual string ErrorAddress
        {
            get { return Endpoint.Metadata.PoisonMessageAddress; }
        }

        public bool IsSendOnly
        {
            get
            {
                if (Endpoint != null && Endpoint.Metadata != null)
                    return Endpoint.Metadata.SendOnly;
                return false;
            }
        }
        public int MaxRetries { get; set; }

        public MessageFormat MessageFormat { get; set; }

        public IMessageSerializer MessageSerializer { get; set; }

        public virtual int NumberOfWorkerThreads {
            get { return ConsumerPools.Default.WorkerCount; }
            set { ConsumerPools.Default.WorkerCount = value; }
        }

        public IObjectBuilder ObjectBuilder { get; set; }

        public virtual string ReturnAddress {
            get {
                if(Endpoint != null && Endpoint.Metadata != null)
                    return Endpoint.Metadata.ReturnAddress;

                return null;
            }
        }
       
        //-----< Public and protected methods >-----
        //----------------------------------------------------------------------------------        

        protected virtual IConnection CreateConnection() {
            var _connection = ConnectionFactory.CreateConnection();
            _connection.ExceptionListener += OnConnectionException;
            _connection.AcknowledgementMode = AcknowledgementMode.Transactional;
            return _connection;
        }

        protected virtual void OnConnectionException(Exception exception) {
            ILog logger = LogManager.GetLogger(typeof(ActiveMQTransport));
            logger.Error("OnConnectionException:" + exception.Message, exception);
        }

        protected static Boolean IsPrimitiveType(Type type) {
            return type == typeof(Int32) ||
                type == typeof(Int64) ||
                type == typeof(String) ||
                type == typeof(Byte) ||
                type == typeof(Boolean) ||
                type == typeof(Double) ||
                type == typeof(Char);
        }

        public virtual ActiveMQTransactionScope CreateScope() {
            EnsureConnection();
            var _scope = fConnection.BeginScope();
            _scope.DestinationResolver = this.DestinationResolver;
            return _scope;
        }

        public virtual MessageEnvelope Ask(MessageEnvelope envelope, String destination, TimeSpan timeout) {
            EnsureConnection();
            using(var _session = fConnection.CreateSession(AcknowledgementMode.ClientAcknowledge)) {
                var _tempQueue = _session.CreateTemporaryQueue();
                try {
                    IMessage _message;
                    using(var _scope = _session.BeginScope()) {
                        try {
                            _scope.DestinationResolver = this.DestinationResolver;
                            envelope.ReturnAddress = _tempQueue.ToString();
                            envelope.Recoverable = false;
                            _message = Send(_scope, envelope, destination, timeout.Add(TimeSpan.FromMinutes(1)));
                            _scope.Commit();
                        }
                        catch {
                            _scope.Rollback();
                            throw;
                        }
                    }
                    using(var _consumer = _session.CreateConsumer(_tempQueue)) {
                        var _response = _consumer.Receive(timeout) as ActiveMQMessage;
                        if(_response != null) {
                            if(_response.NMSCorrelationID == _message.NMSMessageId) {
                                return CreateTransportMessageFromActiveMQMessage(_response);
                            }
                        }
                    }
                }
                finally {
                    _session.DeleteDestination(_tempQueue);
                }
            }
            return null;
        }

        public virtual void Send(MessageEnvelope envelope, params string[] destinations) {
            using(var _scope = CreateScope()) {
                try {
                    Send(_scope, envelope, destinations);
                    _scope.Commit();
                }
                catch {
                    _scope.Rollback();
                    throw;
                }
            }
        }

        public virtual void Send(ActiveMQTransactionScope scope, MessageEnvelope envelope, params string[] destinations) {
            if(destinations == null)
                throw new ArgumentNullException("destinations");
            foreach(var _destination in destinations) {
                Send(scope, envelope, _destination, (TimeSpan?)null);
            }
        }

        protected virtual IMessage Send(ActiveMQTransactionScope scope, MessageEnvelope envelope, String destination, TimeSpan? timeToLive) {
            if(scope == null)
                throw new ArgumentNullException("scope");
            if(envelope == null)
                throw new ArgumentNullException("envelope");
            if(destination == null)
                throw new ArgumentNullException("destination");
            if(this.MessageSerializer == null)
                this.MessageSerializer = Endpoint.LocalServices.GetService<IMessageSerializer>();
            IMessage _message = null;
            var _activeMQEnvelope = envelope as ActiveMQMessageEnvelope;
            if(_activeMQEnvelope != null && _activeMQEnvelope.BytesMessage) {
                _message = scope.Session.CreateBytesMessage(envelope.Payload);
                _message.Properties["BinaryStream"] = "Unknown";
                foreach(var _property in envelope.Body.GetType().GetProperties()) {
                    if(_property.Name != "Bytes" && _property.CanRead &&
                       _property.GetCustomAttributes(typeof(NonSerializedAttribute), true).Length == 0 &&
                       _property.GetCustomAttributes(typeof(XmlIgnoreAttribute), true).Length == 0) {
                        var _value = _property.GetValue(envelope.Body, null);
                        if(_value != null) {
                            if(!IsPrimitiveType(_value.GetType())) {
                                var _enumerable = _value as IEnumerable;
                                if(_enumerable != null) {
                                    var _list = new ArrayList();
                                    foreach(var _item in _enumerable)
                                        _list.Add(_item);
                                    _value = _list;
                                }
                                else if(_value is DateTime)
                                    _value = ((DateTime)_value).ToString("o", CultureInfo.InvariantCulture);
                                else {
                                    var _converter = TypeDescriptor.GetConverter(_value.GetType());
                                    if(_converter != null && _converter.CanConvertTo(typeof(String)))
                                        _value = _converter.ConvertToInvariantString(_value);
                                    else {
                                        var _formattable = _value as IFormattable;
                                        if(_formattable != null)
                                            _value = _formattable.ToString(null, CultureInfo.InvariantCulture);
                                        else
                                            _value = _value.ToString();
                                    }
                                }
                            }
                            _message.Properties[_property.Name] = _value;
                        }
                    }
                }
                _message.Properties["MessageClass"] = envelope.Body.GetType().AssemblyQualifiedName;
            }
            else if(_activeMQEnvelope != null && MessageFormat == MessageFormat.Text && _activeMQEnvelope.TextPayload != null) {
                _message = scope.Session.CreateTextMessage(_activeMQEnvelope.TextPayload);
            }
            else if(MessageFormat == MessageFormat.Text && (this.MessageSerializer is XmlMessageSerializer || this.MessageSerializer is DataContractMessageSerializer)) {
                var _text = Encoding.Default.GetString(envelope.Payload);
                _message = scope.Session.CreateTextMessage(_text);
            }
            else
                _message = scope.Session.CreateBytesMessage(envelope.Payload);
            _message.NMSCorrelationID = envelope.CorrelationID;
            if(envelope.ReturnAddress != null)
                _message.NMSReplyTo = DestinationResolver.Resolve(envelope.ReturnAddress);
            foreach(var _attributeKey in envelope.Headers.AllKeys) {
                if(!fExcludedHeaderProperties.Contains(_attributeKey))
                    _message.Properties[_attributeKey] = envelope.Headers[_attributeKey];
            }
            var _messageGroupIdProperty = envelope.Body.GetType().GetProperty("MessageGroupId");
            if(_messageGroupIdProperty != null) {
                var _messageGroupId = _messageGroupIdProperty.GetValue(envelope.Body, null);
                if(_messageGroupId != null)
                    ((ActiveMQMessage)_message).NMSXGroupID = _messageGroupId.ToString();
            }
            scope.Send(_message, destination, envelope.Recoverable, NMSConstants.defaultPriority, timeToLive);
            return _message;
        }

        public virtual int GetNumberOfPendingMessages() {
            return 0;
        }

        protected virtual void OnFinishedMessageProcessing() {
            var _event = MessageProcessingFinished;
            if(_event != null)
                _event(this, new EventArgs());
        }

        protected virtual bool OnMessageProcessingError(Exception error, MessageEnvelope message) {
            var _event = MessageProcessingError;
            if(_event != null) {
                _event(this, new MessageProcessingErrorEventArgs { Exception = error, Message = message });
                return true;
            }
            return false;
        }

        protected virtual void OnTransportMessageReceived(MessageReceivedEventArgs args) {
            EventHandler<MessageReceivedEventArgs> _event = MessageReceived;
            if(_event != null)
                _event(this, args);
        }

        protected virtual void OnStartedMessageProcessing() {
            var _event = MessageProcessingStarted;
            if(_event != null)
                _event(this, new EventArgs());
        }

        public virtual void ReceiveMessageLater(MessageEnvelope m) {
            Send(m, ReturnAddress);
        }

        protected void EnsureConnection() {
            lock(fConnectionLock) {
                if(fConnection != null && !fConnection.IsStarted)
                    CloseConnection();
                if(fConnection == null) {
                    fConnection = CreateConnection();
                    fConnection.Start();
                    StartConsumerPools();
                }
            }
        }

        protected virtual void StartConsumerPools() {
            foreach(var _pool in fConsumerPools) {
                StartConsumerPool(_pool);
            }
        }
        
        protected virtual void StartConsumerPool(ActiveMQConsumerPool pool) {
            for(Int32 i = 0; i < pool.WorkerCount; i++) {
                if(pool.HasQueues || pool.HasTopics) 
                    StartConsumerPool(pool, i);
            }
        }
        
        protected virtual void StartConsumerPool(ActiveMQConsumerPool pool, Int32 workerIndex) {
            var _session = fConnection.CreateSession(pool.AcknowledgementMode);
            foreach(var _queue in pool.Queues)
                ConsumeQueue(pool, workerIndex, _session, _queue);
            if(workerIndex == 0) {
                // Note that we can't have serveral workers for a topic as they would receive a duplicate of the same message and thus process it n times!
                foreach(var _topic in pool.Topics)
                    ConsumeTopic(pool, workerIndex, _session, _topic);
            }
        }
        
        protected virtual void ConsumeQueue(ActiveMQConsumerPool pool, Int32 workerIndex, ISession session, ActiveMQConsumedQueue queue) {
            var _queue = (ActiveMQDestination)DestinationResolver.Resolve(queue.Name);
            if(queue.IsExclusive)
                _queue.SetExclusive(true);
            var _consumer = session.CreateConsumer(_queue, queue.Selector);
            var _binding = new QueueBinding(pool, workerIndex, session, queue);
            _consumer.Listener += new MessageListener(_binding, this).OnMessage;
        }

        protected virtual void ConsumeTopic(ActiveMQConsumerPool pool, Int32 workerIndex, ISession session, ActiveMQConsumedTopic topic) {

            String _topicName = topic.Name;
            if(!DefaultDestinationResolver.HasDestinationPrefix(_topicName))
                _topicName = DefaultDestinationResolver.TopicPrefix + _topicName;
            var _topic = (ITopic)DestinationResolver.Resolve(_topicName);
            IMessageConsumer _consumer;
            if(topic.DurableSubscriptionName != null)
                _consumer = session.CreateDurableConsumer(_topic, topic.DurableSubscriptionName, topic.Selector, topic.NoLocal);
            else
                _consumer = session.CreateConsumer(_topic, topic.Selector, topic.NoLocal);
            var _binding = new QueueBinding(pool, workerIndex, session, topic);
            _consumer.Listener += new MessageListener(_binding, this).OnMessage;
        }

        protected override void StartService() {
            var _destinationResolver = Endpoint.LocalServices.GetService<IDestinationResolver>();
            if(_destinationResolver == null) {
                _destinationResolver = new DefaultDestinationResolver();
                Endpoint.LocalServices.RegisterService<IDestinationResolver>(_destinationResolver);
            }
            DestinationResolver = _destinationResolver;
            if(ConsumerPools.HasConsumers) {
                EnsureConnection();
            }
        }

        protected internal void ExecuteListener(QueueBinding binding, IMessage message) {
            ActiveMQMessageEnvelope messageEnvelope = null;
            try {

                var _mqMessage = ActiveMQMessage.Transform(message);

                OnStartedMessageProcessing();

                messageEnvelope = (ActiveMQMessageEnvelope)CreateTransportMessageFromActiveMQMessage(_mqMessage);

                if(binding.Queue.Handler != null) {
                    try {
                        binding.Queue.Handler(message);
                    }
                    catch(Exception ex) {
                        messageEnvelope.ProcessingException = new MessageProcessingException(ex.Message, ex, messageEnvelope);
                    }
                }
                else {
                    OnTransportMessageReceived(new MessageReceivedEventArgs(messageEnvelope));
                }
                if(binding.Session.Transacted) {
                    if(messageEnvelope.ProcessingException != null)
                        binding.Session.Rollback();
                    else
                        binding.Session.Commit();
                }
                else if(binding.Session.AcknowledgementMode == AcknowledgementMode.ClientAcknowledge) {
                    if(messageEnvelope.ProcessingException == null) 
                        _mqMessage.Acknowledge();
                }

                OnFinishedMessageProcessing();
            }
            catch(Exception ex) {
                if(binding.Session.Transacted)
                    binding.Session.Rollback();
                //If we are succesfull in notifying someone about this error,
                //we'll go on our merry old way, assuming that it has been dealt with.
                //Otherwise, we'll re-throw, maybe someone down the line can help.
                if(!OnMessageProcessingError(ex, messageEnvelope))
                    throw;

            }
        }

        protected void CloseConnection() {
            if(fConnection != null) {
                lock(fConnectionLock) {
                    if(fConnection != null) {
                        fConnection.ExceptionListener -= OnConnectionException;
                        var _connection = fConnection;
                        fConnection = null;
                        _connection.Dispose();
                    }
                }
            }
        }

        protected override void ShutDownService() {
            CloseConnection();
        }

        public virtual MessageEnvelope CreateEmptyMessageEnvelope() {
            var message = new ActiveMQMessageEnvelope();
            return message;
        }

        protected virtual MessageEnvelope CreateTransportMessageFromActiveMQMessage(ActiveMQMessage mqMessage) {
            var result = (ActiveMQMessageEnvelope)CreateEmptyMessageEnvelope();
            result.TransportMessage = mqMessage;

            foreach(String key in mqMessage.Properties.Keys) {
                var _value = mqMessage.Properties[key];
                if(_value != null)
                    result.Headers.Add(key, _value.ToString());
            }

            result.MessageID = mqMessage.NMSMessageId;
            if(mqMessage.NMSReplyTo != null)
                result.ReturnAddress = mqMessage.NMSReplyTo.ToString();
            result.Recoverable = mqMessage.NMSDeliveryMode == MsgDeliveryMode.Persistent;
            result.CorrelationID = mqMessage.NMSCorrelationID;
            result.RetryCount = mqMessage.NMSXDeliveryCount;

            var _bytesMessage = mqMessage as IBytesMessage;
            if(_bytesMessage != null) {
                result.Payload = _bytesMessage.Content;
                return result;
            }
            var _textMessage = mqMessage as ITextMessage;
            if(_textMessage != null) {
                result.TextPayload = _textMessage.Text;
                return result;
            }
            throw new InvalidOperationException(String.Format(CultureInfo.InvariantCulture, "Unexpected message type {0}", mqMessage.GetType()));
        }


        //-----< Events >-----
        //----------------------------------------------------------------------------------        

        public event EventHandler MessageProcessingStarted;
        public event EventHandler MessageProcessingFinished;
        public event EventHandler<MessageProcessingErrorEventArgs> MessageProcessingError;
        public event EventHandler<MessageReceivedEventArgs> MessageReceived;

        //-----< Internal members >-----
        //----------------------------------------------------------------------------------        


        //-----< Private members >-----
        //----------------------------------------------------------------------------------        


        //-----< All nested types >-----
        //----------------------------------------------------------------------------------        

        protected internal class QueueBinding {
            private ActiveMQConsumerPool fPool;
            private Int32 fWorkerIndex;
            private ISession fSession;
            private ActiveMQConsumedQueue fQueue;

            public QueueBinding(ActiveMQConsumerPool pool, Int32 workerIndex, ISession session, ActiveMQConsumedQueue queue) {
                fPool = pool;
                fWorkerIndex = workerIndex;
                fSession = session;
                fQueue = queue;
            }

            public ActiveMQConsumerPool Pool {
                get { return fPool; }
            }

            public Int32 WorkerIndex {
                get { return fWorkerIndex; }
            }

            public ISession Session {
                get { return fSession; }
            }
            public ActiveMQConsumedQueue Queue {
                get { return fQueue; }
            }

        }

        private class MessageListener {
            private QueueBinding fBinding;
            private ActiveMQTransport fTransport;

            public MessageListener(QueueBinding binding, ActiveMQTransport transport) {
                fBinding = binding;
                fTransport = transport;
            }

            public void OnMessage(IMessage message) {
                fTransport.ExecuteListener(fBinding, message);
            }
        }

        //-----< All explicit interface implementations >-----
        //----------------------------------------------------------------------------------        

        #region IMessageProcessingErrorHandler Members

        public bool HandleMessageProcessingError(ref SimpleServiceBus.Bus.MessageManagement.MessageProcessingException exception, MessageEnvelope message, IMessageBus messageBus) {
            var _message = (ActiveMQMessageEnvelope)message;
            _message.ProcessingException = exception;
            return true;
        }

        #endregion
    }
}

